package org.fle4y.framework.net.aio;

import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;
import org.fle4y.common.utils.configure.Configure;
import org.fle4y.common.utils.log.LogUtil;
import org.fle4y.framework.net.LogicHandler;
import org.fle4y.framework.net.Protocal;
import org.fle4y.framework.net.Server;
import org.fle4y.framework.net.Session;

/**
 * author : xiehui<br>
 * company : northking <br>
 * Create at: 2012-6-8<br>
 * @version : 1.0<br>
 * mailto:flexie@foxmail.com<br>
 *  Description : AIO实现网络服务
 */
public class AIOServer implements Server{
	private static final Logger log = LogUtil.getLogger(AIOServer.class.getName());
	
	/**会话集合*/
	private ConcurrentHashMap<Long,Session> sessionMap=new ConcurrentHashMap<Long,Session>();
	/**配置文件*/
	private Configure configure;
	/**异步服务通道*/
	private AsynchronousServerSocketChannel asynchronousServerSocketChannel;
	
	private AsynchronousChannelGroup  asynchronousChannelGroup;
	
	/**协议*/
	private Protocal protocal ;
	
	/**业务处理*/
	private LogicHandler logic;
	
	/**服务开启还是关闭*/
	private boolean running=false;
	
	/**会话Id产生器*/
	private AtomicLong sessionId=new AtomicLong(0);
	
	private Future<AsynchronousSocketChannel> future;
	
	private  AIOServer(){
		configure = new AIOConfigure();
		try {
			asynchronousChannelGroup=AsynchronousChannelGroup.withCachedThreadPool(
					Executors.newCachedThreadPool(),
					Integer.parseInt(configure.get("pool_size").trim()));
			
		} catch (Exception e) {
			
		}
	}
	public AIOServer(Protocal protocal,LogicHandler logic){
		this();
		this.protocal=protocal;
		this.logic=logic;
		logic.setServer(this);
	}
	public void start() throws Exception {
		log.info("开始准备启动网络通讯服务器...");
		running=true;
		/**创建异步服务通道*/
		asynchronousServerSocketChannel=AsynchronousServerSocketChannel.open(asynchronousChannelGroup);
		
		/**设置TCP属性*/
		asynchronousServerSocketChannel
        .setOption(
                StandardSocketOptions.SO_REUSEADDR,true);
		/**
		 * 绑定地址端口
		 *  其中的100用于指定等待连接的队列大小(backlog)
		 * */
		asynchronousServerSocketChannel=asynchronousServerSocketChannel.bind(new InetSocketAddress("localhost",Integer.parseInt(configure.get("port").trim())),100);
		/**启动监听机器人*/
		log.info("启动监听机器人...");
		while(running){
			pendingAccept();
		}
	}
	public void pendingAccept() {
			/**监听*/
			log.info("开始监听端口: "+ configure.get("port").trim());
				if (running&&asynchronousServerSocketChannel.isOpen()) {
					 future=asynchronousServerSocketChannel.accept();
					 try {
						AsynchronousSocketChannel channel=future.get();
						AIOSession session =new AIOSession(sessionId.incrementAndGet(),channel,protocal,logic);
						sessionMap.put(session.getId(),session);
						session.init(configure);
						session.start();
					} catch (Exception e) {
						e.printStackTrace();
					}
				} else {
			            throw new IllegalStateException("Controller has been closed");
				}
	}
	public void stop() throws Exception {
	}
	
	public Session getSession(long sessionId) {
		return sessionMap.get(sessionId);
	}
	public void setSessionMap(ConcurrentHashMap<Long, Session> sessionMap) {
		this.sessionMap = sessionMap;
	}
	public ConcurrentHashMap<Long, Session> getSessionMap() {
		return sessionMap;
	}
}
